package com.heima.article.listener;

import com.heima.common.constants.message.HotArticleConstants;

import com.heima.common.util.JsonUtils;
import com.heima.model.article.dtos.ArticleBehaviorMsg;
import com.heima.model.article.dtos.ArticleBehaviorStreamMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Serdes;

import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.StringUtils;

import java.time.Duration;

@Slf4j
//@EnableBinding(BehaviorStreamListener.BehaviorStreamProcessor.class)
public class BehaviorStreamListener {

    @StreamListener(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC)
            @SendTo(HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
    public KStream<String, String> process(KStream<String, String> stream){
        //输入: {"articleId":1121212, "type":"VIEW", "num":1 } + {"articleId":1121212, "type":"LIKE", "num":1 }
        //输出: {"articleId":1121212, viewNum:1, likeNum:1, commentNum:0, collectNum:0}
        return stream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                log.info("收到原始数据：{}", value);
                ArticleBehaviorMsg articleBehaviorMsg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                Long articleId = articleBehaviorMsg.getArticleId();
                return new KeyValue<>(""+articleId, value);
            }
        }).groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)).grace(Duration.ZERO))
                .aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        log.info("时间窗口初始化");
                        return null;
                    }
                }, new Aggregator<String, String, String>() {
                    @Override
                    public String apply(String key, String value, String aggregate) {
                        log.info("时间窗口内聚合计算，key：{}，value:{}, aggregate:{}", key, value, aggregate);
                        ArticleBehaviorMsg behaviorMsg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                        if(StringUtils.isEmpty(aggregate)){
                            //  时间窗口内的第一条消息
                            ArticleBehaviorStreamMsg resultMsg= new ArticleBehaviorStreamMsg();
                            resultMsg.setArticleId(behaviorMsg.getArticleId());
                            if(ArticleBehaviorMsg.ArticleBehaviorType.VIEWS.equals(behaviorMsg.getType())){
                                resultMsg.setViews(behaviorMsg.getNum());
                            }else if(ArticleBehaviorMsg.ArticleBehaviorType.LIKES.equals(behaviorMsg.getType())){
                                resultMsg.setLikes(behaviorMsg.getNum());
                            }else if(ArticleBehaviorMsg.ArticleBehaviorType.COMMENT.equals(behaviorMsg.getType())){
                                resultMsg.setComment(behaviorMsg.getNum());
                            }else if(ArticleBehaviorMsg.ArticleBehaviorType.COLLECTION.equals(behaviorMsg.getType())){
                                resultMsg.setCollection(behaviorMsg.getNum());
                            }
                            return JsonUtils.toString(resultMsg);
                        }else{
                            //  时间窗口内消息累加
                            // {"articleId":1121212, "type":"LIKE", "num":1 } +  {"articleId":1121212, viewNum:1, likeNum:1, commentNum:0, collectNum:0}
                            ArticleBehaviorStreamMsg resultMsg = JsonUtils.toBean(aggregate, ArticleBehaviorStreamMsg.class);
                            if(ArticleBehaviorMsg.ArticleBehaviorType.VIEWS.equals(behaviorMsg.getType())){
                                resultMsg.setViews(behaviorMsg.getNum() + (resultMsg.getViews()==null?0:resultMsg.getViews()));
                            }else if(ArticleBehaviorMsg.ArticleBehaviorType.LIKES.equals(behaviorMsg.getType())){
                                resultMsg.setLikes(behaviorMsg.getNum() + (resultMsg.getLikes()==null?0:resultMsg.getLikes()));
                            }else if(ArticleBehaviorMsg.ArticleBehaviorType.COMMENT.equals(behaviorMsg.getType())){
                                resultMsg.setComment(behaviorMsg.getNum() + (resultMsg.getComment()==null?0:resultMsg.getComment()));
                            }else if(ArticleBehaviorMsg.ArticleBehaviorType.COLLECTION.equals(behaviorMsg.getType())){
                                resultMsg.setCollection(behaviorMsg.getNum() + (resultMsg.getCollection()==null?0:resultMsg.getCollection()));
                            }
                            return  JsonUtils.toString(resultMsg);
                        }
                    }
                }, Materialized.with(Serdes.String(), Serdes.String()))
                .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
                .toStream()
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, String value) {
                        log.info("时间窗口结束，key：{}，value:{}, startTime:{}, endTime:{}", key.key(), value, key.window().startTime(), key.window().endTime());
                        return new KeyValue<>(key.key(), value);
                    }
                });
    }

    public interface BehaviorStreamProcessor{
        @Input(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC)
        public KStream<String, String> input();

        @Output(HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
        public KStream<String, String> output();

    }
}